package a.dg;

import a.pojo.StationLog;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//001,181,182,success,1000,40
// 002,182,183,fai1,3000,20
//001,183,184,success,2000,30
// 002,184,185,success,6000,50
// 003,181,183,fai1,5000,50
// 001,183,184,success,3000,10
// 002,184,185,success,4000,40
// 001,181,183,success,7000,50

public class MyDatagenTable {
    public static void main(String[] args) {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
        SingleOutputStreamOperator<StationLog> stationLogSingleOutputStreamOperator = environment.socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] wodds = line.split(",");
                    return new StationLog(wodds[0], wodds[1], wodds[2], wodds[3], Long.valueOf(wodds[4]), Long.valueOf(wodds[5]));
                });

        Table table = tableEnvironment.fromDataStream(stationLogSingleOutputStreamOperator, Schema.newBuilder()
//完全不指定可以自动推断
                .column("sid", DataTypes.STRING())
                .column("call_out", DataTypes.STRING())
                .column("call_in", DataTypes.STRING())
                .column("call_type", DataTypes.STRING())
                .column("call_time", DataTypes.BIGINT())
                .column("duration", DataTypes.BIGINT())
//                获得处理时间
                .columnByExpression("proc_time", "PROCTIME()")
//                计算业务时间
                .columnByExpression("rowtime", "TO_TIMESTAMP_LTZ(call_time,3)")
                .watermark("rowtime", "rowtime - INTERVAL '2' SECOND")
                .build());

        table.printSchema();
        table.execute().print();
    }

    public static void main2(String[] args) {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
        SingleOutputStreamOperator<StationLog> stationLogSingleOutputStreamOperator = environment.socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] wodds = line.split(",");
                    return new StationLog(wodds[0], wodds[1], wodds[2], wodds[3], Long.valueOf(wodds[4]), Long.valueOf(wodds[5]));
                });

        Table table = tableEnvironment.fromDataStream(stationLogSingleOutputStreamOperator);
        table.printSchema();
        table.execute().print();
    }

    public static void main1(String[] args) {
//        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
//        TableEnvironment tEnv = TableEnvironment.create(settings);
//        tEnv.getConfig().set("parallelism", "1");
//
////        DataTypes.TIMESTAMP(3)
//        tEnv.executeSql("CREATE TABLE data_gen (\n" +
//                "    sid  BIGINT,\n" +
//                "    call_out      BIGINT,\n" +
//                "    call_in      BIGINT,\n" +
//                "    call_type      BIGINT,\n" +
//
//                "    transaction_time TIMESTAMP(3),\n" +
//                "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '3' SECOND\n" +
//                ") WITH (\n" +
//                "    'connector' = 'datagen'," +
//                "    'rows-per-second' = '1'," +
//                "    'number-of-rows' = '3'," +
//                "    'fields.sid.kind' = 'random'," +
//                "    'fields.sid.min' = '1'," +
//                "    'fields.sid.max' = '3'," +
//                "    'fields.call_out.kind' = 'random'," +
//                "    'fields.call_out.min' = '100'," +
//                "    'fields.call_out.max' = '200'" +
//                "    'fields.call_in.kind' = 'random'," +
//                "    'fields.call_in.min' = '100'," +
//                "    'fields.call_in.max' = '200'" +
//                "    'fields.call_type.kind' = 'random'," +
//                "    'fields.call_type.min' = '0'," +
//                "    'fields.call_type.max' = '1'" +
//
//                ")");
//
//
//        tEnv.executeSql("CREATE TABLE print (\n" +
//                "    account_id BIGINT,\n" +
//                "    log_ts     TIMESTAMP(3),\n" +
//                "    amount     BIGINT\n," +
//                "    PRIMARY KEY (account_id, log_ts) NOT ENFORCED" +
//                ") WITH (\n" +
//                "   'connector'  = 'print'\n" +
//                ")");
//
//        Table data_gen = tEnv.from("data_gen");
//        Table report = report(data_gen);
//        report.executeInsert("print");
    }
}
